package com.dsk.connector.source.common;

import cn.sanenen.sunutils.queue.SMQ;
import com.dsk.connector.commons.conf.TableMapping;
import com.dsk.connector.commons.conf.TableRegMapping;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;

import java.util.List;
import java.util.Map;

public abstract class DebeziumHandler<E> implements DebeziumEngine.ChangeConsumer<E> {

    private static Logger logger= LoggerFactory.getLogger(DebeziumHandler.class);

    @Value("${config.source.regTableName:false}")
    protected boolean regTableName;

    protected Map<String, TableMapping> tableMappings;
    protected Map<String, TableRegMapping> tableRegMappings;

    @Autowired
    protected Environment environment;

    /**
     * 处理接收到的数据.<br>
     * 此方法不可抛出异常，否则将会导致binlog处理中断，变成僵尸程序
     *
     * @param records
     * @param committer
     * @throws InterruptedException
     */
    @Override
    public void handleBatch(List<E> records, DebeziumEngine.RecordCommitter<E> committer) throws InterruptedException {
        int batchSize = records.size();
        int processedCount = 0;
        for (E event : records) {
            try{
                processedCount++;
                handleRecord(event,processedCount,batchSize);
            }catch (Throwable ex){
                ex.printStackTrace();
                logger.error("消费binglog异常{}",ex);
            }finally {
                committer.markProcessed(event);
            }
        }
        committer.markBatchFinished();
    }

    protected abstract void handleRecord(E event,int processedCount,int batchSize) throws Exception;

    public void sendMessage(String topic,String message){
        SMQ.push(topic,message);
    }

    public Map<String, TableMapping> getTableMappings() {
        return tableMappings;
    }

    public void setTableMappings(Map<String, TableMapping> tableMappings) {
        this.tableMappings = tableMappings;
    }

    public Map<String, TableRegMapping> getTableRegMappings() {
        return tableRegMappings;
    }

    public void setTableRegMappings(Map<String, TableRegMapping> tableRegMappings) {
        this.tableRegMappings = tableRegMappings;
    }

    public boolean isRegTableName() {
        return regTableName;
    }

    public void setRegTableName(boolean regTableName) {
        this.regTableName = regTableName;
    }

    public void setEnvironment(Environment environment) {
        this.environment=environment;
    }
}
